package com.example.demo.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
@ConditionalOnProperty(name = "kafka.enabled", havingValue = "true", matchIfMissing = true) // havingValue true时启动  matchIfMiss 默认true
public class BigDataTopicListener {

    private static final Logger log = LoggerFactory.getLogger(BigDataTopicListener.class);

    /**
     * 监听kafka数据（批量消费）
     * @param consumerRecords
     * @param ack
     */
    @KafkaListener(topics = "mykafka", containerFactory = "batchFactory")
    public void batchConsumer(List<ConsumerRecord<?, ?>> consumerRecords, Acknowledgment ack) {
        long start = System.currentTimeMillis();

            try {
                consumerRecords.forEach(record -> {

                        // 获取消息内容
                        String key = (String) record.key();
                        String value = (String) record.value();
                        long offset = record.offset();
                        int partition = record.partition();
                        log.info("处理消息: 分区={}, 偏移量={}, 键={}, 值={}", partition, offset, key, value);


                        // 3 db save

                        // 4. 手动提交偏移量（所有消息处理成功后）
                        ack.acknowledge();
                });
            } catch (Exception e) {
                // 5. 异常处理（根据业务需求选择回滚或跳过）
                log.error("处理消息批次时发生异常", e);

                // 示例：失败后重试策略（可根据业务调整）
                // 1. 直接抛异常，让消息重新消费（需确保幂等性）
                // throw new RuntimeException("消息处理失败，触发重试", e);

                // 2. 或者记录失败消息，继续处理其他消息
                // saveFailedMessages(consumerRecords);
                // ack.acknowledge(); // 跳过失败消息，继续消费后续消息

            }


        log.info("收到bigData推送的数据，拉取数据量：{}，消费时间：{}ms", consumerRecords.size(), (System.currentTimeMillis() - start));
    }

}